package io.kiki.sba.registry.server.data_store.remoting.sessionserver.handler;


import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.dataserver.DatumVersion;
import io.kiki.sba.registry.common.model.dataserver.GetDataVersionRequest;
import io.kiki.sba.registry.common.model.slot.SlotAccess;
import io.kiki.sba.registry.common.model.slot.SlotAccessGenericResponse;
import io.kiki.sba.registry.common.model.store.StringInterner;
import io.kiki.sba.registry.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

import static io.kiki.sba.registry.server.data_store.remoting.sessionserver.handler.HandlerMetrics.GetVersion.GET_VERSION_COUNTER;

/**
 * processor to get versions of specific dataInfoIds
 */
public class GetDataVersionsHandler extends AbstractDataHandler<GetDataVersionRequest> {
    private static final Logger logger = LoggerFactory.getLogger(GetDataVersionsHandler.class);

    @Autowired
    private ThreadPoolExecutor getDataProcessorExecutor;

    @Override
    public Executor getExecutor() {
        return getDataProcessorExecutor;
    }

    @Override
    public void checkParam(GetDataVersionRequest request) {
        StringUtil.checkNonNegative(request.getSlotId(), "GetDataVersionRequest.slotId");
        StringUtil.checkNotBlank(request.getDataCenter(), "GetDataVersionRequest.dataCenter");
        checkClientInterfaceId(request.getClientInterfaceId());
    }

    @Override
    public Object doHandle(Channel channel, GetDataVersionRequest request) {
        processClientInterfaceId(channel, request.getClientInterfaceId());
        final int slotId = request.getSlotId();
        final String dataCenter = request.getDataCenter();
        final SlotAccess slotAccessBefore = checkSlotAccess(dataCenter, slotId, request.getSlotsEpoch(), request.getSlotLeaderEpoch());
        if (!slotAccessBefore.isAccept()) {
            return SlotAccessGenericResponse.failedResponse(slotAccessBefore);
        }
        final Map<String, DatumVersion> requestDataInfoIdToDatumVersionMap = request.getDataInfoIdToDatumVersionMap();
        Map<String /*dataInfoId*/, DatumVersion> dataInfoIdToDatumVersionMap = datumStorageDelegate.getVersions(dataCenter, slotId, requestDataInfoIdToDatumVersionMap.keySet());
        // double check slot access, @see GetDataHandler
        final SlotAccess slotAccessAfter = checkSlotAccess(dataCenter, slotId, request.getSlotsEpoch(), request.getSlotLeaderEpoch());
        if (slotAccessAfter.getSlotLeaderEpoch() != slotAccessBefore.getSlotLeaderEpoch()) {
            return SlotAccessGenericResponse.failedResponse(slotAccessAfter, "slotLeaderEpoch has change, prev=" + slotAccessBefore);
        }
        Map<String, DatumVersion> map = new HashMap<>(64);
        for (Map.Entry<String, DatumVersion> entry : requestDataInfoIdToDatumVersionMap.entrySet()) {
            final String dataInfoId = entry.getKey();
            final DatumVersion requstDatumVersion = entry.getValue();
            final DatumVersion datumVersion = dataInfoIdToDatumVersionMap.get(dataInfoId);
            // contains the datum which is interested
            if (datumVersion != null) {
                if (requstDatumVersion.getValue() < datumVersion.getValue()) {
                    // need to notify session
                    map.put(dataInfoId, datumVersion);
                } else
                    if (requstDatumVersion.getValue() > datumVersion.getValue()) {
                        // the session.push version is bigger than datum.version. this may happens:
                        // 1. slot-1 own by data-A, balance slot-1, migrating from data-A to data-B.
                        //    this need a very short time window to broadcast the information. e.g. a heartbeat
                        // interval time
                        // 2.1. part of session/data has not updated the Slots. [session-A, data-A]
                        // 2.2. another part of the session/data has updated the Slots. [session-B, data-B]
                        // 3. data-B finish migrating and session-B put publisher-B to data-B gen a new
                        // datum.version=V1
                        // 4. session-A put publisher-A to data-A, gen a new datum.version=V2(V1<V2), and push V2
                        // to subscriber-A
                        // 5. session-A update the Slots, connect to data-B, but the
                        // subscriber-A(push.version)=V2 is
                        //    bigger than current datum.version=V1, the publisher-B would not push to
                        // subscriber-A.
                        // so, we need to compare the push.version and datum.version
                        DatumVersion updateVer = datumStorageDelegate.updateVersion(dataCenter, dataInfoId);
                        map.put(dataInfoId, updateVer);
                        logger.info("updateV,{},{},{},interestVer={},currentVer={},updateVer={}", slotId, dataInfoId, dataCenter, requstDatumVersion, datumVersion, updateVer);
                    }
                // if equals, do not return the version to reduce network overhead
            } else {
                // todo xiaojian.xj
                // no datum in data node, this maybe happens an empty datum occurs migrating
                // there is subscriber subs the dataId. we create a empty datum to trace the version
                // the version will trigger the push after session.scan
                // cache the dataInfoId
                final String cacheDataInfoId = StringInterner.intern(dataInfoId);
                final DatumVersion emptyDatumIfAbsent = datumStorageDelegate.createEmptyDatumIfAbsent(dataCenter, cacheDataInfoId);
                if (emptyDatumIfAbsent != null) {
                    map.put(dataInfoId, emptyDatumIfAbsent);
                }
                logger.info("createV,{},{},{},interestVer={},createV={}", slotId, dataInfoId, dataCenter, requstDatumVersion, emptyDatumIfAbsent);
            }
        }
        logger.info("getV,{},{},gets={},rets={}", slotId, dataCenter, dataInfoIdToDatumVersionMap.size(), map.size());
        GET_VERSION_COUNTER.inc();
        return SlotAccessGenericResponse.successResponse(slotAccessAfter, map);
    }

    @Override
    public Class interest() {
        return GetDataVersionRequest.class;
    }
}
